package com.game.gateway.channel;


import com.game.common.cloud.GameChannelCloseEvent;
import com.game.common.concurrent.GameEventExecutorGroup;
import com.game.gateway.rpc.GameRpcService;
import com.game.netty.game.common.IGameMessage;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import java.util.HashMap;
import java.util.Map;

/**
 *
 * @ClassName: GameMessageEventDispatchService
 * @Description: GameChannel的管理类，主要负责管理用户id与GameChannel的映射关系，一个玩家始终只有一个GameChannel,并负责请求消息的分发
 * @author: wgs
 * @date: 2019年5月30日 下午8:31:25
 */
public class GameMessageEventDispatchService {
    private static Logger logger = LoggerFactory.getLogger(GameMessageEventDispatchService.class);
    private Map<Long, GameChannel> gameChannelGroup = new HashMap<>();// 管理PlayerId与GameChannel的集合

    private GameEventExecutorGroup workerGroup;// 业务处理线程池组
    private EventExecutor executor;// 当前管理gameChannelGroup集合的事件线程池
    private IMessageSendFactory messageSendFactory; // 向客户端发送消息的接口类，可以根据需求，有不同的实现，这里默认是发送到kafka的消息总线服务中。
    private GameChannelInitializer channelInitializer;
    private GameRpcService gameRpcSendFactory;
    private ApplicationContext context;

    public GameMessageEventDispatchService(ApplicationContext context, GameEventExecutorGroup workerGroup, IMessageSendFactory messageSendFactory, GameRpcService gameRpcSendFactory, GameChannelInitializer channelInitializer) {
        this.executor = workerGroup.next();
        this.workerGroup = workerGroup;
        this.messageSendFactory = messageSendFactory;
        this.channelInitializer = channelInitializer;
        this.gameRpcSendFactory = gameRpcSendFactory;
        this.context = context;
    }

    public ApplicationContext getApplicationContext() {
        return context;
    }
    // 此方法保证所有操作gameChannelGroup集合的行为都在同一个线程中执行，避免跨线程操作。
    private void safeExecute(Runnable task) {// 将方法的请求变成事件，在此类所属的事件线程池中执行
        if (this.executor.inEventLoop()) {// 如果当前调用这个方法的线程和此类所属的线程是同一个线程，则可以立刻执行执行。
            try {
                task.run();
            } catch (Throwable e) {
                logger.error("服务器内部错误", e);
            }
        } else {
            this.executor.execute(() -> {// 如果当前调用这个方法的线程和此类所属的线程不是同一个线程，将此任务提交到线程池中等待执行。
                try {
                    task.run();
                } catch (Throwable e) {
                    logger.error("服务器内部错误", e);
                }
            });
        }
    }

    private GameChannel getGameChannel(Long playerId) {
        GameChannel gameChannel = this.gameChannelGroup.get(playerId);
        if (gameChannel == null) {// 从集合中获取一个GameChannel，如果这个GameChannel为空，则重新创建，并初始化注册这个Channel，完成GameChannel的初始化。
            gameChannel = new GameChannel(playerId, this, messageSendFactory, gameRpcSendFactory);
            this.gameChannelGroup.put(playerId, gameChannel);
            this.channelInitializer.initChannel(gameChannel);// 初始化Channel，可以通过这个接口向GameChannel中添加处理消息的Handler.
            gameChannel.register(workerGroup.select(playerId), playerId);// 发注册GameChannel的事件。
        }
        return gameChannel;
    }

    public void fireReadMessage(Long playerId, IGameMessage message) {// 发送接收到的消息事件
        this.safeExecute(() -> {
            GameChannel gameChannel = this.getGameChannel(playerId);
            gameChannel.fireReadGameMessage(message);
        });
    }

    public void fireReadRPCRequest(IGameMessage gameMessage) {
        this.safeExecute(() -> {
            GameChannel gameChannel = this.getGameChannel(gameMessage.getHeader().getPlayerId());
            gameChannel.fireChannelReadRPCRequest(gameMessage);
        });
    }

    public void fireUserEvent(Long playerId, Object msg, Promise<Object> promise) {// 发送用户定义的事件
        this.safeExecute(() -> {
            GameChannel gameChannel = this.getGameChannel(playerId);
            gameChannel.fireUserEvent(msg, promise);
        });
    }

    public void fireInactiveChannel(Long playerId) {// 发送GameChannel失效的事件，在这个事件中可以处理一些数据落地的操作
        this.safeExecute(() -> {
            GameChannel gameChannel = this.gameChannelGroup.remove(playerId);
            if (gameChannel != null) {
                gameChannel.fireChannelInactive();
                // 发布GameChannel失效事件
                GameChannelCloseEvent event = new GameChannelCloseEvent(this, playerId);
                context.publishEvent(event);
            }
        });
    }

    public void broadcastMessage(IGameMessage gameMessage, long... playerIds) {// 发送消息广播事件，客多个客户端发送消息。
        if (playerIds == null || playerIds.length == 0) {
            logger.debug("广播的对象集合为空，直接返回");
            return;
        }
        this.safeExecute(() -> {
            for (long playerId : playerIds) {
                if (this.gameChannelGroup.containsKey(playerId)) {
                    GameChannel gameChannel = this.getGameChannel(playerId);
                    gameChannel.pushMessage(gameMessage);
                }
            }
        });
    }

    public void broadcastMessage(IGameMessage gameMessage) {
        this.safeExecute(() -> {
            this.gameChannelGroup.values().forEach(channel -> {
                channel.pushMessage(gameMessage);
            });
        });
    }


}
